Federated Learningを大規模に実運用しようと思った際に重要なオブジェクトとなる、Planという概念について紹介します。Planは使用するネットワーク帯域を劇的に減らし、非同期処理を実現し、リモートデイバイスに自律性を与えてくれます。元となるアイデアはこの論文、Towards Federated Learning at Scale: System Design、を参照してください。現在はPySyftライブラリのニーズに応じて、一部変更が加えられています。
Planは、関数のように、連続するオペレーションを纏める目的で作られています。しかし、Planを使えば、定義した一連のオペレーションを一回のメッセージでリモートのワーカーに送ることができます。こうすることで、N個の（オペレーションの）メッセージを送る代わりに1つのメッセージを送るだけで、ポインタを通してN個のオペレーションを参照できます。PlanにはTensor("state tensors"と呼ばれます)をつけて送ることもできます。"state tensors"は引数のようなものです。Planは送信可能な関数と捉えることもできますし、リモートにて実行可能なクラスと捉えることもできます。これによって、高次のユーザーはPlanの概念を特に意識することなく、恣意的な連続するPyTorchの関数をリモートワーカーに送ることが可能になります。
一点注意が必要な点は、現時点ではPlanで使用可能な関数はPyTorchのHook機能を持つオペレーションに限定されています。これは if, for そしrて while といった論理構造が使えないことを意味します。私たちはこの件について対応中です。
正確には、これらのオペレーションを使うことはできますが、最初のコンピューテーションで取った分岐がその後の全てのコンピューテーションに適応されてしまいます。これでは都合が悪いですよね。
Authors:
In [ ]:
import torch
import torch.nn as nn
import torch.nn.functional as F
次にPySyft用のコードです。一つ覚えておくべきことは、ローカルワーカーはクライアントワーカーになるべきではないという事です。 クライアンワーカー以外はPlanの実行に必要なオブジェクトを保持できません。ここで言うローカルワーカーとは私たちで、クライアントワーカーとはリモートワーカーの事です。
In [ ]:
import syft as sy # Pysyftライブラリをインポート
hook = sy.TorchHook(torch) # PyTorchをホック ie torchを拡張します
# IMPORTANT: ローカルワーカーはクライアントワーカーになることは出来ません
hook.local_worker.is_client_worker = False
server = hook.local_worker
説明の通り、リモートワーカー（デバイス）を定義します。 そして、彼らにデータを割り当てます。
In [ ]:
x11 = torch.tensor([-1, 2.]).tag('input_data')
x12 = torch.tensor([1, -2.]).tag('input_data2')
x21 = torch.tensor([-1, 2.]).tag('input_data')
x22 = torch.tensor([1, -2.]).tag('input_data2')
device_1 = sy.VirtualWorker(hook, id="device_1", data=(x11, x12))
device_2 = sy.VirtualWorker(hook, id="device_2", data=(x21, x22))
devices = device_1, device_2
In [ ]:
@sy.func2plan()
def plan_double_abs(x):
x = x + x
x = torch.abs(x)
return x
Planが作成できました。チェックしてみましょう。
In [ ]:
plan_double_abs
In [ ]:
pointer_to_data = device_1.search('input_data')[0]
pointer_to_data
もし、ここでlocation:device_1のデバイス上でPlayを実行しようとすると、エラーになってしまいます。まだビルドが出来ていないからです。
In [ ]:
plan_double_abs.is_built
In [ ]:
# ビルドされていないPlanをリモートワーカーへ送ろうとするとエラーになります
try:
plan_double_abs.send(device_1)
except RuntimeError as error:
print(error)
Planをビルドするには、必要な引数（何らかのデータ）を渡しつつbuildコマンドを実行してください。Planがビルドされると全てのコマンドはローカルワーカーによって、順番に実行され、結果はPlayのactions属性にキャッシュされます。
In [ ]:
plan_double_abs.build(torch.tensor([1., -2.]))
In [ ]:
plan_double_abs.is_built
このPlayを再度送ってみましょう。今度はうまく行きます。
In [ ]:
# 今回はエラーは出ません
pointer_plan = plan_double_abs.send(device_1)
pointer_plan
Tensorの時と同様にポインタが取得できます。PointerPlanという名前は分かり易いですね。
特筆すべきことの一つは、Planはビルドされる時にコンピューテーションの結果として割り当てられるIDが事前に設定されます。これにより、リモートマシンのコンピューテーションの結果を待たずしてIDを取得でき、コマンドは非同期で送信が可能になります。例えば、device_1でのバッチ処理の結果を待たずにdevice_2で次のバッチ処理を実行することが可能になります。
In [ ]:
pointer_to_result = pointer_plan(pointer_to_data)
print(pointer_to_result)
そして、計算された値は、今までと同様の手法で受け取ることが可能です。
In [ ]:
pointer_to_result.get()
In [ ]:
class Net(sy.Plan):
def __init__(self):
super(Net, self).__init__()
self.fc1 = nn.Linear(2, 3)
self.fc2 = nn.Linear(3, 2)
def forward(self, x):
x = F.relu(self.fc1(x))
x = self.fc2(x)
return F.log_softmax(x, dim=0)
In [ ]:
net = Net()
In [ ]:
net
ダミーデータを使ってビルドしてみましょう。
In [ ]:
net.build(torch.tensor([1., 2.]))
次に、Planをリモートワーカーへ送ってみましょう
In [ ]:
pointer_to_net = net.send(device_1)
pointer_to_net
リモートデータのポインタを取得しましょう
In [ ]:
pointer_to_data = device_1.search('input_data')[0]
構文的にはリモートマシン上で逐次的にオペレーションを実行するのと何らかわりはありません。ですが、この手法では複数のオペレーションが一回のコミュニケーションで実行されています。
In [ ]:
pointer_to_result = pointer_to_net(pointer_to_data)
pointer_to_result
データの受け取りはいつも通りです。
In [ ]:
pointer_to_result.get()
ジャジャーン！ローカルワーカー（サーバー、この場合は私たち？）とリモートデバイスの間のコミュニケーションを劇的に減らすことに成功しました！
In [ ]:
class Net(sy.Plan):
def __init__(self):
super(Net, self).__init__()
self.fc1 = nn.Linear(2, 3)
self.fc2 = nn.Linear(3, 2)
def forward(self, x):
x = F.relu(self.fc1(x))
x = self.fc2(x)
return F.log_softmax(x, dim=0)
In [ ]:
net = Net()
# Planのビルド
net.build(torch.tensor([1., 2.]))
メインのステップです。
In [ ]:
pointer_to_net_1 = net.send(device_1)
pointer_to_data = device_1.search('input_data')[0]
pointer_to_result = pointer_to_net_1(pointer_to_data)
pointer_to_result.get()
実は、同じPlanから別のPointerPlansをビルドすることが可能です。別のデバイスでPlanをリモート実行する時と同じです。
In [ ]:
pointer_to_net_2 = net.send(device_2)
pointer_to_data = device_2.search('input_data')[0]
pointer_to_result = pointer_to_net_2(pointer_to_data)
pointer_to_result.get()
注記: この例では、Planは一つのオペレーションしか実行しています。実行されていたのはforwardです。
In [ ]:
@sy.func2plan(args_shape=[(-1, 1)])
def plan_double_abs(x):
x = x + x
x = torch.abs(x)
return x
plan_double_abs.is_built
args_shapeはPlanをビルドする際にダミーデータを作成するのに使われます。
In [ ]:
@sy.func2plan(args_shape=[(1, 2), (-1, 2)])
def plan_sum_abs(x, y):
s = x + y
return torch.abs(s)
plan_sum_abs.is_built
state引数を使って実データを渡すことも可能です。
In [ ]:
@sy.func2plan(args_shape=[(1,)], state=(torch.tensor([1]), ))
def plan_abs(x, state):
bias, = state.read()
x = x.abs()
return x + bias
In [ ]:
pointer_plan = plan_abs.send(device_1)
x_ptr = torch.tensor([-1, 0]).send(device_1)
p = pointer_plan(x_ptr)
p.get()
もっと知りたい方はチュートリアルPart 8 bisを参照してください。
一番簡単に貢献できる方法はこのGitHubのレポジトリにスターを付けていただくことです。スターが増えると露出が増え、より多くのデベロッパーにこのクールな技術の事を知って貰えます。
最新の開発状況のトラッキングする一番良い方法はSlackに入ることです。 下記フォームから入る事ができます。 http://slack.openmined.org
コミュニティに貢献する一番良い方法はソースコードのコントリビューターになることです。PySyftのGitHubへアクセスしてIssueのページを開き、"Projects"で検索してみてください。参加し得るプロジェクトの状況を把握することができます。また、"good first issue"とマークされているIssueを探す事でミニプロジェクトを探すこともできます。
もし、ソースコードで貢献できるほどの時間は取れないけど、是非何かサポートしたいという場合は、寄付をしていただくことも可能です。寄附金の全ては、ハッカソンやミートアップの開催といった、コミュニティ運営経費として利用されます。
In [ ]: